/**
 * (C) Copyright 2019-2021 Intel Corporation.
 *
 * SPDX-License-Identifier: BSD-2-Clause-Patent
 */
/**
  * Implementation for aggregation and discard
 */
#define D_LOGFAC	DD_FAC(vos)

#include <daos_srv/vos.h>
#include <daos/object.h>	/* for daos_unit_oid_compare() */
#include <daos/checksum.h>
#include <daos_srv/srv_csum.h>
#include "vos_internal.h"
#include "evt_priv.h"

/*
 * EV tree sorted iterator returns logical entry in extent start order, and
 * the information like: physical entry it belongs to, visibility, is it the
 * last entry, etc. will be returned as well. One thing worth noting is that
 * sorted iterator is working on an out-of-band auxiliary index, so it won't
 * be affected by any in-tree modifications.
 *
 * EV tree aggregation is driven by sorted iterator: As the iterator moves on,
 * the visible logical entries will be queued to form a merge window, and all
 * the physical entries (no matter if it's fully covered or not) within the
 * window will be queued too. When the window size reaches certain threshold,
 * a procedure called merge window flush will be triggered to replace the old
 * physical entries by the new coalesced physical entries.
 *
 * On merge window flush, the queued visible logical entries will be used to
 * facilitate the data transfer from old physical entries to new coalesced
 * physical entries. If any old physical entry (not fully covered in current
 * window) straddles window end, it has to be head-truncated on window flush,
 * and the remaining part will be processed in next merge window.
 */

/* EV tree physical entry */
struct agg_phy_ent {
	d_list_t		pe_link;
	/* Original in-tree rectangle */
	struct evt_rect		pe_rect;
	/* Entry payload address */
	bio_addr_t		pe_addr;
	/* Original csums for entry */
	struct dcs_csum_info	pe_csum_info;
	/*
	 * Extent start offset for the truncated physical entry. (Entry
	 * straddles window end could be truncated on merge window flush)
	 */
	daos_off_t		pe_off;
	/* Pool map version of entry */
	uint32_t		pe_ver;
	/* Referenced by visible entries */
	uint32_t		pe_ref;
	/* Need to truncate on window flush */
	bool			pe_trunc_head;
	/** Mark the entry as removed */
	bool			pe_remove;
};

/* Removal record */
struct agg_rmv_ent {
	d_list_t		re_link;
	/** list link for physical records */
	d_list_t		re_phy_link;
	/* In tree rectangle */
	struct evt_rect		re_rect;
	/** Real entries, if any, contained in a logical rectangle */
	d_list_t		re_contained;
	int			re_aggregate:1, /* Aggregate of one or more records */
				re_child:1;	/* Contained in aggregate record */
	/** Refcount of physical records that reference this removal */
	int			re_phy_count;
};

/* EV tree logical entry */
struct agg_lgc_ent {
	struct evt_extent	 le_ext;
	struct agg_phy_ent	*le_phy_ent;
};

/*
 * EV tree logical segment, it'll be used to form new physical
 * rectangle and being inserted in evtree on merge window flush.
 */
struct agg_lgc_seg {
	/* Start index in mw_lgc_ents */
	unsigned int		 ls_idx_start;
	/* End index in mw_lgc_ents */
	unsigned int		 ls_idx_end;
	/* New segment generated by truncate on window flush */
	struct agg_phy_ent	*ls_phy_ent;
	/* Description of the new physical entry to be inserted */
	struct evt_entry_in	 ls_ent_in;
};

/* I/O context used on EV tree merge window flush */
struct agg_io_context {
	/* Temporary buffer for data transfer */
	void			*ic_buf;
	unsigned int		 ic_buf_len;
	/* Buffer to hold output csums for entire aggregation */
	void			*ic_csum_buf;
	unsigned int		 ic_csum_buf_len;
	/* Array of structs used for recalculation of checksums */
	struct csum_recalc	*ic_csum_recalcs;
	unsigned int		 ic_csum_recalc_cnt;
	/* Segments being involved on merge window flush */
	struct agg_lgc_seg	*ic_segs;
	unsigned int		 ic_seg_max;
	unsigned int		 ic_seg_cnt;
	/* Reserved SCM extents for new physical entries */
	struct vos_rsrvd_scm	*ic_rsrvd_scm;
	/* Reserved NVMe extents for new physical entries */
	d_list_t		 ic_nvme_exts;
	void			 (*ic_csum_recalc_func)(void *);
};

/* Merge window for evtree aggregation */
struct agg_merge_window {
	/* Record size */
	daos_size_t			 mw_rsize;
	/* Threshold for merge window flush */
	daos_size_t			 mw_flush_thresh;
	/* Merge window extent */
	struct evt_extent		 mw_ext;
	/** Real merge window upper bound */
	uint64_t			 mw_alloc_hi;
	/* Physical entries in merge window */
	d_list_t			 mw_phy_ents;
	unsigned int			 mw_phy_cnt;
	/** Possibly deleted physical entries */
	d_list_t			 mw_rmv_ents;
	/** In order list of physical removal records */
	d_list_t			 mw_phy_rmv_ents;
	unsigned int			 mw_rmv_cnt;
	/* Visible logical entries in merge window */
	struct agg_lgc_ent		*mw_lgc_ents;
	unsigned int			 mw_lgc_max;
	unsigned int			 mw_lgc_cnt;
	/* I/O context for transferring data on flush */
	struct agg_io_context		 mw_io_ctxt;
	bool				 mw_csum_support;
};

struct vos_agg_param {
	uint32_t		ap_credits_max; /* # of tight loops to yield */
	uint32_t		ap_credits;	/* # of tight loops */
	daos_handle_t		ap_coh;		/* container handle */
	daos_unit_oid_t		ap_oid;		/* current object ID */
	daos_key_t		ap_dkey;	/* current dkey */
	daos_key_t		ap_akey;	/* current akey */
	unsigned int		ap_discard:1,
				ap_csum_err:1,
				ap_full_scan:1,
				ap_discard_obj:1;
	struct umem_instance	*ap_umm;
	bool			(*ap_yield_func)(void *arg);
	void			*ap_yield_arg;
	/* SV tree: Max epoch in specified iterate epoch range */
	daos_epoch_t		 ap_max_epoch;
	/* EV tree: Merge window for evtree aggregation */
	struct agg_merge_window	 ap_window;
	bool			 ap_skip_akey;
	bool			 ap_skip_dkey;
	bool			 ap_skip_obj;
};

static int
agg_del_entry(daos_handle_t ih, struct umem_instance *umm,
	      vos_iter_entry_t *entry, unsigned int *acts)
{
	int	rc;

	D_ASSERT(umm != NULL);
	D_ASSERT(acts != NULL);

	rc = umem_tx_begin(umm, NULL);
	if (rc)
		return rc;

	rc = vos_iter_delete(ih, NULL);
	if (rc != 0)
		rc = umem_tx_abort(umm, rc);
	else
		rc = umem_tx_commit(umm);

	if (rc) {
		D_ERROR("Failed to delete entry: "DF_RC"\n", DP_RC(rc));
		return rc;
	}

	*acts |= VOS_ITER_CB_DELETE;
	return rc;
}

static inline void
reset_agg_pos(vos_iter_type_t type, struct vos_agg_param *agg_param)
{
	switch (type) {
	case VOS_ITER_OBJ:
		memset(&agg_param->ap_oid, 0, sizeof(agg_param->ap_oid));
		break;
	case VOS_ITER_DKEY:
		memset(&agg_param->ap_dkey, 0, sizeof(agg_param->ap_dkey));
		break;
	case VOS_ITER_AKEY:
		memset(&agg_param->ap_akey, 0, sizeof(agg_param->ap_akey));
		break;
	default:
		break;
	}
}

static inline bool
need_aggregate(struct vos_agg_param *agg_param, vos_iter_entry_t *entry)
{
	struct vos_container	*cont = vos_hdl2cont(agg_param->ap_coh);

	/** Skip this check for discard */
	if (agg_param->ap_discard_obj || agg_param->ap_discard)
		return true;

	D_DEBUG(DB_EPC, "full_scan:%d, hae:"DF_U64", last_update:"DF_U64", "
		"flags:%u\n", agg_param->ap_full_scan,
		cont->vc_cont_df->cd_hae, entry->ie_last_update,
		entry->ie_vis_flags);

	/* Don't skip aggregation for full scan */
	if (agg_param->ap_full_scan)
		return true;

	/* Don't skip aggregation when the obj/dkey/akey is punched */
	if (entry->ie_vis_flags & VOS_VIS_FLAG_COVERED)
		return true;

	D_ASSERT(entry->ie_last_update != 0);
	return entry->ie_last_update >= cont->vc_cont_df->cd_hae;
}

static int
vos_agg_obj(daos_handle_t ih, vos_iter_entry_t *entry,
	    struct vos_agg_param *agg_param, unsigned int *acts)
{
	D_ASSERT(agg_param != NULL);
	if (daos_unit_oid_compare(agg_param->ap_oid, entry->ie_oid)) {
		if (need_aggregate(agg_param, entry)) {
			D_DEBUG(DB_EPC, "oid:"DF_UOID" vos agg starting\n",
				DP_UOID(entry->ie_oid));
			agg_param->ap_oid = entry->ie_oid;
			reset_agg_pos(VOS_ITER_DKEY, agg_param);
			reset_agg_pos(VOS_ITER_AKEY, agg_param);
		} else {
			D_DEBUG(DB_EPC, "Skip untouched oid:"DF_UOID"\n",
				DP_UOID(agg_param->ap_oid));
			*acts |= VOS_ITER_CB_SKIP;
		}
	} else {
		/*
		 * When recursive vos_iterate() yield in sub tree, re-probe
		 * is required when it returns back to upper level tree, if
		 * the just processed object is found on re-probe, we need
		 * to notify vos_iterate() to not iterate into to sub tree
		 * again.
		 */
		D_DEBUG(DB_EPC, "Skip oid:"DF_UOID" aggregation on re-probe\n",
			DP_UOID(agg_param->ap_oid));
		*acts |= VOS_ITER_CB_SKIP;
	}

	return 0;
}

static inline int
vos_agg_key_compare(daos_key_t key1, daos_key_t key2)
{
	if (key1.iov_len != key2.iov_len)
		return 1;

	return memcmp(key1.iov_buf, key2.iov_buf, key1.iov_len);
}

static int
vos_agg_dkey(daos_handle_t ih, vos_iter_entry_t *entry,
	     struct vos_agg_param *agg_param, unsigned int *acts)
{
	D_ASSERT(agg_param != NULL);
	if (vos_agg_key_compare(agg_param->ap_dkey, entry->ie_key)) {
		if (need_aggregate(agg_param, entry)) {
			agg_param->ap_dkey = entry->ie_key;
			reset_agg_pos(VOS_ITER_AKEY, agg_param);
		} else {
			D_DEBUG(DB_EPC, "Skip untouched dkey: "DF_KEY"\n",
				DP_KEY(&entry->ie_key));
			*acts |= VOS_ITER_CB_SKIP;
		}
	} else {
		D_DEBUG(DB_EPC, "Skip dkey: "DF_KEY" aggregation on re-probe\n",
			DP_KEY(&entry->ie_key));
		*acts |= VOS_ITER_CB_SKIP;
	}

	return 0;
}

static inline bool
ext1_covers_ext2(struct evt_extent *ext1, struct evt_extent *ext2)
{
	D_ASSERT(ext1->ex_lo <= ext1->ex_hi);
	D_ASSERT(ext2->ex_lo <= ext2->ex_hi);

	return (ext1->ex_lo <= ext2->ex_lo && ext1->ex_hi >= ext2->ex_hi);
}

enum {
	MW_CLOSED	= 0,	/* Window closed, resource released */
	MW_FLUSHED,		/* Window flushed, no logical entries */
	MW_OPENED,		/* Window opened, has logical entries */
};

static int
merge_window_status(struct agg_merge_window *mw)
{
	struct agg_io_context	*io = &mw->mw_io_ctxt;

	D_ASSERT(io->ic_seg_cnt == 0);
	D_ASSERT(io->ic_rsrvd_scm == NULL ||
		 io->ic_rsrvd_scm->rs_actv_at == 0);
	D_ASSERT(d_list_empty(&io->ic_nvme_exts));

	D_ASSERT(mw->mw_ext.ex_lo <= mw->mw_ext.ex_hi);

	if (mw->mw_lgc_cnt != 0) {
		D_ASSERT(mw->mw_rsize != 0);
		D_ASSERT(mw->mw_phy_cnt != 0);
		D_ASSERT(!d_list_empty(&mw->mw_phy_ents));

		return MW_OPENED;
	}

	D_ASSERT(mw->mw_ext.ex_lo == 0 && mw->mw_ext.ex_hi == 0);

	if (mw->mw_lgc_ents != NULL) {
		/*
		 * Even there if there isn't any logical entries,
		 * there could be some truncated physical entries.
		 */
		D_ASSERT(mw->mw_rsize != 0);
		return MW_FLUSHED;
	}

	if (mw->mw_phy_cnt != 0 && mw->mw_rmv_cnt != 0) {
		/* There are physical records that are partially covered by removal records */
		return MW_FLUSHED;
	}

	/* Window closed, all resource should have been released */
	D_ASSERT(mw->mw_phy_cnt == 0);
	D_ASSERT(d_list_empty(&mw->mw_phy_ents));
	D_ASSERT(mw->mw_lgc_max == 0);

	D_ASSERT(io->ic_buf_len == 0);
	D_ASSERT(io->ic_buf == NULL);
	D_ASSERT(io->ic_seg_max == 0);
	D_ASSERT(io->ic_segs == NULL);
	D_ASSERT(io->ic_rsrvd_scm == NULL);

	return MW_CLOSED;
}

static int
vos_agg_akey(daos_handle_t ih, vos_iter_entry_t *entry,
	     struct vos_agg_param *agg_param, unsigned int *acts)
{
	D_ASSERT(agg_param != NULL);
	if (vos_agg_key_compare(agg_param->ap_akey, entry->ie_key)) {
		if (need_aggregate(agg_param, entry)) {
			agg_param->ap_akey = entry->ie_key;
		} else {
			D_DEBUG(DB_EPC, "Skip untouched akey: "DF_KEY"\n",
				DP_KEY(&entry->ie_key));
			*acts |= VOS_ITER_CB_SKIP;
		}
	} else {
		D_DEBUG(DB_EPC, "Skip akey: "DF_KEY" aggregation on re-probe\n",
			DP_KEY(&entry->ie_key));
		*acts |= VOS_ITER_CB_SKIP;
	}

	if (agg_param->ap_discard) {
		/* No merge window for discard path so bypass checks below. */
		return 0;
	}

	/* Reset the max epoch for low-level SV tree iteration */
	agg_param->ap_max_epoch = 0;
	/* The merge window for EV tree aggregation should have been closed */
	if (merge_window_status(&agg_param->ap_window) != MW_CLOSED)
		D_ASSERTF(false, "Merge window isn't closed.\n");

	/* Reset the output checksum buffer, since all overlap consumed. */
	if (agg_param->ap_window.mw_csum_support) {
		D_FREE(agg_param->ap_window.mw_io_ctxt.ic_csum_buf);
		agg_param->ap_window.mw_io_ctxt.ic_csum_buf_len = 0;
	}

	return 0;
}

static int
vos_agg_sv(daos_handle_t ih, vos_iter_entry_t *entry,
	   struct vos_agg_param *agg_param, unsigned int *acts)
{
	int	rc;

	D_ASSERT(agg_param != NULL);
	D_ASSERT(entry->ie_epoch != 0);

	/* Discard */
	if (agg_param->ap_discard)
		goto delete;

	/* If entry is covered, the key or object is punched */
	if (entry->ie_vis_flags & VOS_VIS_FLAG_COVERED)
		goto delete;
	/*
	 * Aggregate: preserve the first recx which has highest epoch, because
	 * of re-probe, the highest epoch could be iterated multiple times.
	 */
	if (agg_param->ap_max_epoch == 0 ||
	    agg_param->ap_max_epoch == entry->ie_epoch) {

		switch (entry->ie_dtx_state) {
		case DTX_ST_COMMITTED:
			/* Highest epoch is committed, keep it. */
			agg_param->ap_max_epoch = entry->ie_epoch;
			break;
		case DTX_ST_PREPARED:
			/*
			 * Highest epoch is uncommitted.  Since it may be
			 * punched by a key or object and that entity may not
			 * know about the update, we need to abort processing
			 * of the current single value for now.
			 */
			D_DEBUG(DB_EPC, "Hit uncommitted single value at epoch:"
				DF_X64"\n", entry->ie_epoch);
			return -DER_TX_BUSY;
		case DTX_ST_ABORTED:
			/*
			 * Highest epoch is aborted, delete it and continue
			 * checking on next lower epoch.
			 */
			D_DEBUG(DB_EPC, "Delete aborted at epoch:"DF_X64"\n",
				entry->ie_epoch);
			goto delete;
		default:
			D_ASSERTF(0, "Unexpected DTX state: %d\n",
				  entry->ie_dtx_state);
			break;
		}

		return 0;
	}

	D_ASSERTF(entry->ie_epoch < agg_param->ap_max_epoch,
		  "max:"DF_U64", cur:"DF_U64"\n",
		  agg_param->ap_max_epoch, entry->ie_epoch);

delete:
	rc = agg_del_entry(ih, agg_param->ap_umm, entry, acts);
	if (rc) {
		D_ERROR("Failed to delete SV entry: "DF_RC"\n", DP_RC(rc));
	} else if (vos_iter_empty(ih) == 1 && agg_param->ap_discard) {
		/* Trigger re-probe in akey iteration */
		*acts |= VOS_ITER_CB_YIELD;
	}

	return rc;
}

/* Allocates sub-ranges of the checksum buffer to each output segment. */
static unsigned int
csum_prepare_ent(struct evt_entry_in *ent_in, unsigned int cs_type,
		 unsigned int cs_len, unsigned int chunksize)
{
	unsigned int cur_cnt = csum_chunk_count(chunksize,
						ent_in->ei_rect.rc_ex.ex_lo,
						ent_in->ei_rect.rc_ex.ex_hi,
						ent_in->ei_inob);

	ent_in->ei_csum.cs_nr = cur_cnt;
	ent_in->ei_csum.cs_type = cs_type;
	ent_in->ei_csum.cs_len = cs_len;
	ent_in->ei_csum.cs_buf_len = cur_cnt * ent_in->ei_csum.cs_len;
	ent_in->ei_csum.cs_chunksize = chunksize;

	return cur_cnt * ent_in->ei_csum.cs_len;
}

/* Each new segment requires an allocated buffer range to hold the checksums
 * calculated for the new segment. This buffer range is also used to hold
 * the verification checksum for the component (input) segments.
 * The full buffer is extended to hold checksums for entire merge window.
 * Currently, allocations for prior windows are retained until aggregation
 * for an evtree is complete (in vos_agg_akey, and at end of agggregation).
 */
static int
csum_prepare_buf(struct agg_lgc_seg *segs, unsigned int seg_cnt,
		 void **csum_bufp, unsigned int cur_len, unsigned int new_len)
{
	unsigned char	*buffer = NULL;
	unsigned int	 cur_buf = 0;
	int		 i;

	if (new_len > cur_len) {
		D_REALLOC_NZ(buffer, *csum_bufp, new_len);
		if (buffer == NULL)
			return -DER_NOMEM;
	} else
		buffer = *csum_bufp;

	memset(buffer, 0, new_len);
	for (i = 0; i < seg_cnt; i++) {
		struct dcs_csum_info *csum_info = &segs[i].ls_ent_in.ei_csum;

		csum_info->cs_csum = &buffer[cur_buf];
		cur_buf += csum_info->cs_len * csum_info->cs_nr;
		D_ASSERT(cur_buf <= new_len);
	}
	*csum_bufp = buffer;

	return 0;
}

static inline struct agg_rmv_ent *
allocate_rmv_ent(const struct evt_extent *ext, daos_epoch_t epoch, uint16_t minor_epc)
{
	struct agg_rmv_ent *rm_ent;

	D_ALLOC_PTR(rm_ent);
	if (rm_ent == NULL)
		return NULL;

	D_INIT_LIST_HEAD(&rm_ent->re_contained);
	rm_ent->re_rect.rc_ex.ex_lo = ext->ex_lo;
	rm_ent->re_rect.rc_ex.ex_hi = ext->ex_hi;
	rm_ent->re_rect.rc_minor_epc = minor_epc;
	rm_ent->re_rect.rc_epc = epoch;

	return rm_ent;
}

static inline void
recx2ext(const daos_recx_t *recx, struct evt_extent *ext)
{
	D_ASSERT(recx->rx_nr > 0);
	ext->ex_lo = recx->rx_idx;
	ext->ex_hi = recx->rx_idx + recx->rx_nr - 1;
}

static inline int
delete_evt_entry(struct vos_obj_iter *oiter, const vos_iter_entry_t *entry,
		 const char *desc)
{
	struct evt_rect	rect;
	int		rc;

	recx2ext(&entry->ie_orig_recx, &rect.rc_ex);
	rect.rc_epc = entry->ie_epoch;
	rect.rc_minor_epc = entry->ie_minor_epc;

	rc = evt_delete(oiter->it_hdl, &rect, NULL);
	if (rc)
		D_ERROR("Delete %s EV entry "DF_RECT" error: "DF_RC"\n",
			desc, DP_RECT(&rect), DP_RC(rc));
	return rc;
}

static int
delete_removal(struct agg_merge_window *mw, struct vos_obj_iter *oiter, struct agg_rmv_ent *rm_ent)
{
	int			 rc = 0;

	D_ASSERT(d_list_empty(&rm_ent->re_contained));
	D_DEBUG(DB_EPC, "Removing physical removal record: "DF_RECT"\n", DP_RECT(&rm_ent->re_rect));
	rc = evt_delete(oiter->it_hdl, &rm_ent->re_rect, NULL);
	if (rc) {
		D_ERROR("Remove "DF_RECT" error: "DF_RC"\n", DP_RECT(&rm_ent->re_rect), DP_RC(rc));
		return rc;
	}

	d_list_del(&rm_ent->re_phy_link);
	d_list_del(&rm_ent->re_link);
	if (!rm_ent->re_child) {
		D_ASSERT(mw->mw_rmv_cnt > 0);
		mw->mw_rmv_cnt--;
	}
	D_FREE(rm_ent);

	return 0;
}

static int
enqueue_rmv_ent(struct agg_merge_window *mw, const struct evt_extent *ext,
		const vos_iter_entry_t *entry, struct vos_obj_iter *oiter)
{
	struct agg_rmv_ent	*rm_ent, *rm_ent2, *rm_ent3;
	d_list_t		*list = &mw->mw_rmv_ents;
	int			 rc;

	/** Remove any past removal records that have no remaining physical references */
	d_list_for_each_entry_safe(rm_ent, rm_ent2, &mw->mw_phy_rmv_ents, re_phy_link) {
		if (rm_ent->re_phy_count != 0)
			continue;

		if (rm_ent->re_rect.rc_ex.ex_hi >= ext->ex_lo)
			continue;

		rc = delete_removal(mw, oiter, rm_ent);

		if (rc != 0)
			return rc;
	}

	rm_ent = allocate_rmv_ent(ext, entry->ie_epoch, entry->ie_minor_epc);
	if (rm_ent == NULL)
		return -DER_NOMEM;

	d_list_for_each_entry_reverse(rm_ent2, &mw->mw_rmv_ents, re_link) {
		if (rm_ent->re_rect.rc_epc != rm_ent2->re_rect.rc_epc)
			continue;

		if ((ext->ex_lo) != rm_ent2->re_rect.rc_ex.ex_hi + 1)
			continue;

		D_DEBUG(DB_EPC, "Removal record "DF_RECT" is contiguous with "DF_RECT"\n",
			DP_RECT(&rm_ent->re_rect), DP_RECT(&rm_ent2->re_rect));

		if (!rm_ent2->re_aggregate) {
			D_ASSERT(d_list_empty(&rm_ent2->re_contained));
			/* Duplicate the entry */
			rm_ent3 = allocate_rmv_ent(&rm_ent2->re_rect.rc_ex, rm_ent2->re_rect.rc_epc,
						   rm_ent2->re_rect.rc_minor_epc);
			if (rm_ent3 == NULL) {
				D_FREE(rm_ent);
				return -DER_NOMEM;
			}
			D_DEBUG(DB_EPC, "Removal record "DF_RECT" duplicated\n",
				DP_RECT(&rm_ent2->re_rect));
			/** Replace the existing entry in the list with the duplicate */
			d_list_add(&rm_ent3->re_link, &rm_ent2->re_link);
			d_list_del(&rm_ent2->re_link);
			/** Add existing entry to duplicate contained list */
			d_list_add_tail(&rm_ent2->re_link, &rm_ent3->re_contained);
			/** Subsequent modification should be on the duplicate */
			rm_ent3->re_aggregate = 1;
			rm_ent2->re_child = 1;
			rm_ent2 = rm_ent3;
		}

		rm_ent2->re_rect.rc_ex.ex_hi = ext->ex_hi;
		list = &rm_ent2->re_contained;
		rm_ent->re_child = 1;
		goto enqueue;
	}
	mw->mw_rmv_cnt++;
enqueue:
	d_list_add_tail(&rm_ent->re_link, list);
	/** Keep a list of just the physical records */
	d_list_add_tail(&rm_ent->re_phy_link, &mw->mw_phy_rmv_ents);

	return 0;
}

static inline bool
phy_ent_is_removed(struct agg_merge_window *mw, const struct evt_extent *phy_ext,
		   daos_epoch_t epoch)
{
	struct agg_rmv_ent	*rm_ent;

	d_list_for_each_entry(rm_ent, &mw->mw_rmv_ents,
			      re_link) {
		struct evt_rect	*rm_rect = &rm_ent->re_rect;

		if (rm_rect->rc_epc == epoch &&
		    rm_rect->rc_ex.ex_lo <= phy_ext->ex_hi &&
		    rm_rect->rc_ex.ex_hi >= phy_ext->ex_hi)
			return true;
	}

	return false;
}

static int
prepare_segments(struct agg_merge_window *mw)
{
	struct agg_io_context	*io = &mw->mw_io_ctxt;
	struct agg_phy_ent	*phy_ent = NULL;
	struct agg_phy_ent	*first = NULL;
	struct agg_phy_ent	*temp = NULL;
	struct agg_lgc_ent	*lgc_ent;
	struct agg_lgc_seg	*lgc_seg;
	struct evt_entry_in	*ent_in;
	struct evt_extent	 ext;
	unsigned int		 i, seg_max;
	unsigned int		 cs_type = 0;
	unsigned int		 cs_len = 0;
	unsigned int		 chunksize = 0;
	unsigned int		 cs_total = 0;
	bool			 hole = false, coalesce;
	int			 rc = 0;

	/*
	 * Allocate large enough segments array to hold all the coalesced
	 * segments (at most mw_lgc_cnt) and truncated segments (at most
	 * mw_phy_cnt).
	 */
	D_ASSERT(mw->mw_rmv_cnt > 0 || (mw->mw_phy_cnt > 0 && mw->mw_lgc_cnt > 0));
	io->ic_seg_cnt = 0;
	if (mw->mw_lgc_cnt == 0)
		goto process_physical;

	seg_max = MAX((mw->mw_lgc_cnt + mw->mw_phy_cnt), 200);
	if (io->ic_seg_max < seg_max) {
		D_REALLOC_ARRAY_NZ(lgc_seg, io->ic_segs, seg_max);
		if (lgc_seg == NULL)
			return -DER_NOMEM;

		io->ic_segs = lgc_seg;
		io->ic_seg_max = seg_max;
	}
	memset(io->ic_segs, 0, io->ic_seg_max * sizeof(*lgc_seg));

	/* Generate coalesced segments according to visible logical entries */
	for (i = 0; i < mw->mw_lgc_cnt; i++) {
		lgc_ent = &mw->mw_lgc_ents[i];
		phy_ent = lgc_ent->le_phy_ent;

		ext = lgc_ent->le_ext;
		D_ASSERT(ext1_covers_ext2(&mw->mw_ext, &ext));

		if (i == 0) {
			coalesce = false;
		} else if (hole != bio_addr_is_hole(&phy_ent->pe_addr)) {
			coalesce = false;
			io->ic_seg_cnt++;
			D_ASSERT(io->ic_seg_cnt < io->ic_seg_max);
		} else {
			coalesce = true;
		}

		hole = bio_addr_is_hole(&phy_ent->pe_addr);
		lgc_seg = &io->ic_segs[io->ic_seg_cnt];
		ent_in = &lgc_seg->ls_ent_in;

		if (!coalesce) {
			lgc_seg->ls_phy_ent = NULL;
			lgc_seg->ls_idx_start = i;
			ent_in->ei_inob = mw->mw_rsize;
			ent_in->ei_rect.rc_ex.ex_lo = ext.ex_lo;
			bio_addr_set_hole(&ent_in->ei_addr, hole);
			if (hole) {
				bio_addr_set(&ent_in->ei_addr, DAOS_MEDIA_SCM,
					     0);
				ent_in->ei_inob = 0;
			}
		} else {
			D_ASSERT(ext.ex_lo == ent_in->ei_rect.rc_ex.ex_hi + 1);
		}

		lgc_seg->ls_idx_end = i;
		ent_in->ei_rect.rc_ex.ex_hi = ext.ex_hi;
		/* Merge to lowest epoch */
		if (ent_in->ei_rect.rc_epc == 0 ||
		    ent_in->ei_rect.rc_epc > phy_ent->pe_rect.rc_epc)
			ent_in->ei_rect.rc_epc = phy_ent->pe_rect.rc_epc;
		/* Merge to lowest pool map version */
		if (ent_in->ei_ver == 0 ||
		    ent_in->ei_ver > phy_ent->pe_ver)
			ent_in->ei_ver = phy_ent->pe_ver;
		ent_in->ei_rect.rc_minor_epc = VOS_SUB_OP_MAX;
	}

	io->ic_seg_cnt++;
	D_ASSERT(io->ic_seg_cnt < io->ic_seg_max);

process_physical:
	/* Generate truncated segments according to physical entries */
	d_list_for_each_entry_safe(phy_ent, temp, &mw->mw_phy_ents, pe_link) {
		if (first == NULL)
			first = phy_ent; /* Save the first one */

		lgc_seg = &io->ic_segs[io->ic_seg_cnt];
		ent_in = &lgc_seg->ls_ent_in;

		ext = phy_ent->pe_rect.rc_ex;

		/* The physical entry was truncated on prev window flush */
		if (phy_ent->pe_off != 0)
			ext.ex_lo += phy_ent->pe_off;

		D_ASSERT(ext.ex_lo <= ext.ex_hi);
		phy_ent->pe_remove = false;
		if (ext.ex_hi > mw->mw_ext.ex_hi) {
			if (phy_ent_is_removed(mw, &ext, phy_ent->pe_rect.rc_epc)) {
				/** If a record is covered by a removal record and is
				 *  contained in the current merge window, it will be
				 *  removed by aggregation algorithm.  If it extends
				 *  into next window, and the tail is fully covered
				 *  by a removal entry, we need to mark the record for
				 *  removal.
				 */
				phy_ent->pe_remove = true;
			}
		}

		/*
		 * Physical entry is in window, or it's fully covered (not
		 * visible) in current window, or the tail (outside of current
		 * window) is fully covered by a removal record.
		 */
		if (ext.ex_hi <= mw->mw_ext.ex_hi || phy_ent->pe_ref == 0 ||
		    phy_ent->pe_remove)
			continue;

		D_ASSERT(ext.ex_lo <= mw->mw_ext.ex_hi);
		D_ASSERT(ext.ex_hi >= mw->mw_ext.ex_lo);

		lgc_seg->ls_phy_ent = phy_ent;
		lgc_seg->ls_idx_start = 0;
		lgc_seg->ls_idx_end = 0;

		ent_in->ei_inob = mw->mw_rsize;
		ent_in->ei_rect.rc_ex.ex_lo = mw->mw_ext.ex_hi + 1;
		ent_in->ei_rect.rc_ex.ex_hi = ext.ex_hi;
		ent_in->ei_rect.rc_epc = phy_ent->pe_rect.rc_epc;
		ent_in->ei_rect.rc_minor_epc = phy_ent->pe_rect.rc_minor_epc;
		ent_in->ei_ver = phy_ent->pe_ver;

		hole = bio_addr_is_hole(&phy_ent->pe_addr);
		bio_addr_set_hole(&ent_in->ei_addr, hole);
		if (hole) {
			bio_addr_set(&ent_in->ei_addr, DAOS_MEDIA_SCM, 0);
			ent_in->ei_inob = 0;
		}

		io->ic_seg_cnt++;
		D_ASSERT(io->ic_seg_cnt <= io->ic_seg_max);
	}
	if (mw->mw_csum_support && io->ic_seg_cnt > 0) {
		D_ASSERT(first != NULL);
		cs_len = first->pe_csum_info.cs_len;
		cs_type = first->pe_csum_info.cs_type;
		chunksize = first->pe_csum_info.cs_chunksize;

		for (i = 0; i < io->ic_seg_cnt; i++) {
			lgc_seg = &io->ic_segs[i];
			ent_in = &lgc_seg->ls_ent_in;
			if (ent_in->ei_inob != 0)
				/* Allocates csum buffer range. */
				cs_total += csum_prepare_ent(ent_in, cs_type,
							     cs_len, chunksize);
		}
		/* Reallocates csum buffer. */
		if (cs_total) {
			rc = csum_prepare_buf(io->ic_segs, io->ic_seg_cnt,
					      &io->ic_csum_buf,
					      io->ic_csum_buf_len, cs_total);
			if (cs_total > io->ic_csum_buf_len)
				io->ic_csum_buf_len = cs_total;
		}
	}
	return rc;
}

static int
reserve_segment(struct vos_object *obj, struct agg_io_context *io,
		daos_size_t size, bio_addr_t *addr)
{
	uint64_t	off;
	uint16_t	media;
	int		rc;

	memset(addr, 0, sizeof(*addr));
	media = vos_media_select(vos_obj2pool(obj), DAOS_IOD_ARRAY, size);

	if (media == DAOS_MEDIA_SCM) {
		off = vos_reserve_scm(obj->obj_cont, io->ic_rsrvd_scm, size);
		if (UMOFF_IS_NULL(off)) {
			D_ERROR("Reserve "DF_U64" from SCM failed.\n", size);
			return -DER_NOSPACE;
		}
		bio_addr_set(addr, media, off);
		return 0;
	}

	D_ASSERT(media == DAOS_MEDIA_NVME);
	rc = vos_reserve_blocks(obj->obj_cont, &io->ic_nvme_exts, size,
				VOS_IOS_AGGREGATION, &off);
	if (rc)
		D_ERROR("Reserve "DF_U64" from NVMe failed. "DF_RC"\n",
			size, DP_RC(rc));
	else
		bio_addr_set(addr, media, off);

	return rc;
}

static inline daos_size_t
merge_window_size(struct agg_merge_window *mw)
{
	struct evt_extent ext;
	D_ASSERT(mw->mw_ext.ex_hi >= mw->mw_ext.ex_lo);
	D_ASSERT(mw->mw_alloc_hi >= mw->mw_ext.ex_lo);
	D_ASSERT(mw->mw_rsize != 0);
	ext.ex_hi = mw->mw_alloc_hi;
	ext.ex_lo = mw->mw_ext.ex_lo;
	return evt_extent_width(&ext) * mw->mw_rsize;
}

/* Widen biov entry for read extents to range required to verify checksums. */
static unsigned int
csum_widen_biov(struct bio_iov *biov, struct agg_phy_ent *phy_ent,
		    struct evt_extent *ext, uint32_t rsize, daos_off_t phy_lo,
		    uint32_t *wider)
{
	struct evt_entry	ent;
	struct evt_extent	aligned_extent = { 0 };
	unsigned int		added_segs = 0;

	ent.en_ext = phy_ent->pe_rect.rc_ex;
	if (phy_lo)
		ent.en_ext.ex_lo = phy_lo;
	ent.en_sel_ext = *ext;
	ent.en_csum = phy_ent->pe_csum_info;
	aligned_extent = evt_entry_align_to_csum_chunk(&ent, rsize);
	bio_iov_set_extra(biov,
			  (ent.en_sel_ext.ex_lo - aligned_extent.ex_lo) *
			  rsize,
			  (aligned_extent.ex_hi - ent.en_sel_ext.ex_hi) *
			  rsize);
	/*Amount to add to IO buffer. */
	*wider = biov->bi_prefix_len + biov->bi_suffix_len;
	added_segs += biov->bi_prefix_len != 0;
	added_segs += biov->bi_suffix_len != 0;
	/* Number of additional read segments for this component extent. */
	return added_segs;
}


/* Extends bio_sglist to include extension to csum boumdaries (added to the end
 * of the list) with added_segs additional entries. These entries will hold the
 * extended prefix and suffix data ranges needed to widen the aggregatable range
 * to prior checksum boundaries.
 */
static int
csum_append_added_segs(struct bio_sglist *bsgl, unsigned int added_segs)
{
	void		*buffer;
	unsigned int	 i, add_idx = bsgl->bs_nr;

	D_REALLOC_ARRAY(buffer, bsgl->bs_iovs, bsgl->bs_nr,
			bsgl->bs_nr + added_segs);
	if (buffer == NULL)
		return -DER_NOMEM;
	bsgl->bs_iovs = buffer;

	for (i = 0; i < bsgl->bs_nr; i++) {
		if (bsgl->bs_iovs[i].bi_prefix_len) {
			/* Add the prefix. */
			D_ASSERT(add_idx < bsgl->bs_nr + added_segs);
			bsgl->bs_iovs[add_idx].bi_addr.ba_off =
					bsgl->bs_iovs[i].bi_addr.ba_off;
			bsgl->bs_iovs[add_idx].bi_data_len =
					bsgl->bs_iovs[i].bi_prefix_len;

			bsgl->bs_iovs[add_idx].bi_addr.ba_type =
				bsgl->bs_iovs[i].bi_addr.ba_type;

			bsgl->bs_iovs[add_idx].bi_prefix_len = 0;
			bsgl->bs_iovs[add_idx].bi_suffix_len = 0;
			bsgl->bs_iovs[add_idx].bi_buf = NULL;
			BIO_ADDR_SET_NOT_HOLE(
				&bsgl->bs_iovs[add_idx++].bi_addr);
		}
		if (bsgl->bs_iovs[i].bi_suffix_len) {
			/* Add the suffix. */
			D_ASSERT(add_idx < bsgl->bs_nr + added_segs);
			bsgl->bs_iovs[add_idx].bi_addr.ba_off =
					bsgl->bs_iovs[i].bi_addr.ba_off +
					bsgl->bs_iovs[i].bi_data_len -
					bsgl->bs_iovs[i].bi_suffix_len;
			bsgl->bs_iovs[add_idx].bi_data_len =
					bsgl->bs_iovs[i].bi_suffix_len;
			bsgl->bs_iovs[add_idx].bi_addr.ba_type =
				bsgl->bs_iovs[i].bi_addr.ba_type;
			bsgl->bs_iovs[add_idx].bi_prefix_len = 0;
			bsgl->bs_iovs[add_idx].bi_suffix_len = 0;
			bsgl->bs_iovs[add_idx].bi_buf = NULL;
			BIO_ADDR_SET_NOT_HOLE(
				&bsgl->bs_iovs[add_idx++].bi_addr);
		}

		/* Reset the parameters for the write (non-extended) data. */

		if (bsgl->bs_iovs[i].bi_prefix_len) {
			bsgl->bs_iovs[i].bi_addr.ba_off +=
						bsgl->bs_iovs[i].bi_prefix_len;
			bsgl->bs_iovs[i].bi_data_len -=
						bsgl->bs_iovs[i].bi_prefix_len;
			bsgl->bs_iovs[i].bi_prefix_len = 0;
		}
		if (bsgl->bs_iovs[i].bi_suffix_len) {
			bsgl->bs_iovs[i].bi_data_len -=
						bsgl->bs_iovs[i].bi_suffix_len;
			bsgl->bs_iovs[i].bi_suffix_len = 0;
		}
	}
	bsgl->bs_nr += added_segs;
	return 0;
}

/* An array of csum_recalc structures is constructed for each output entry.
 * This data is used for checksum verification of the input data, and for
 * calculating the checksum(s) for the output extent.
 */
static void
csum_add_recalcs(struct csum_recalc **recalcs_p,
		 struct agg_phy_ent *phy_ent, struct evt_extent *ext,
		 struct bio_sglist *bsgl, unsigned int idx)
{
	struct csum_recalc      *recalcs = *recalcs_p;

	recalcs[idx].cr_log_ext         = *ext;
	recalcs[idx].cr_phy_ext		= &phy_ent->pe_rect.rc_ex;
	recalcs[idx].cr_phy_ent		= phy_ent;  /* used in this file only */
	recalcs[idx].cr_phy_csum	= &phy_ent->pe_csum_info;
	recalcs[idx].cr_phy_off		= phy_ent->pe_off;
	recalcs[idx].cr_prefix_len	= bsgl->bs_iovs[idx].bi_prefix_len;
	recalcs[idx].cr_suffix_len	= bsgl->bs_iovs[idx].bi_suffix_len;
}

/* Checksum calculations are performed using a call-back function (function
 * pointer passed in the invocation of the aggregation. A single argument
 * struct, passed as a void pointer, is passed to the callback.
 */
static int
csum_recalc(struct agg_io_context *io, struct bio_sglist *bsgl,
	    d_sg_list_t *sgl, struct evt_entry_in *ent_in,
	    struct csum_recalc *recalcs,
	    unsigned int recalc_seg_cnt, daos_size_t seg_size)
{
	struct csum_recalc_args	args = { 0 };

	args.cra_bsgl		= bsgl;
	args.cra_sgl		= sgl;
	args.cra_ent_in		= ent_in;
	args.cra_recalcs	= recalcs;
	args.cra_seg_size	= seg_size;
	args.cra_seg_cnt	= recalc_seg_cnt;
	args.cra_buf		= io->ic_buf;
	args.cra_buf_len	= io->ic_buf_len;

	io->ic_csum_recalc_func(&args);
	if (args.cra_rc == -DER_CSUM && args.cra_bio_ctxt != NULL)
		bio_log_csum_err(args.cra_bio_ctxt, args.cra_tgt_id);
	return args.cra_rc;
}

static int
fill_one_segment(daos_handle_t ih, struct agg_merge_window *mw,
		 struct agg_lgc_seg *lgc_seg, unsigned int *acts)
{
	struct vos_obj_iter	*oiter = vos_hdl2oiter(ih);
	struct vos_object	*obj = oiter->it_obj;
	struct agg_io_context	*io = &mw->mw_io_ctxt;
	struct evt_entry_in	*ent_in = &lgc_seg->ls_ent_in;
	struct agg_phy_ent	*phy_ent;
	struct bio_io_context	*bio_ctxt;
	struct bio_sglist	 bsgl;
	d_sg_list_t		 sgl;
	d_iov_t			 iov;
	bio_addr_t		 addr_dst, addr_src;
	daos_size_t		 seg_size, copy_size, buf_max;
	struct evt_extent	 ext = { 0 };
	daos_off_t		 phy_lo = 0;
	unsigned int		 i, seg_count, biov_idx = 0;
	size_t			 buf_add = 0;
	unsigned int		 added_csum_segs = 0;
	int			 rc;

	D_ASSERT(obj != NULL);
	D_ASSERT(mw->mw_rsize > 0);

	if (bio_addr_is_hole(&ent_in->ei_addr))
		return 0;

	seg_size = evt_rect_width(&ent_in->ei_rect) * mw->mw_rsize;
	D_ASSERTF(seg_size > 0, "seg_size:"DF_U64"\n", seg_size);

	buf_max = MAX(seg_size, merge_window_size(mw));
	buf_max = MAX(buf_max, VOS_MW_FLUSH_THRESH);

	/* Copy data from old logical entries into new segment */
	D_ASSERT(lgc_seg->ls_idx_start <= lgc_seg->ls_idx_end);
	D_ASSERT(lgc_seg->ls_idx_end < mw->mw_lgc_cnt);

	bio_ctxt = obj->obj_cont->vc_pool->vp_io_ctxt;
	D_ASSERT(bio_ctxt != NULL);

	seg_count = lgc_seg->ls_idx_end - lgc_seg->ls_idx_start + 1;
	rc = bio_sgl_init(&bsgl, seg_count);
	if (rc) {
		D_ERROR("Init bsgl error: "DF_RC"\n", DP_RC(rc));
		return rc;
	}

	if (mw->mw_csum_support && seg_count > io->ic_csum_recalc_cnt) {
		void *buffer;

		/* An array of recalc structs (one per output segment). */
		D_REALLOC_ARRAY(buffer, io->ic_csum_recalcs,
				io->ic_csum_recalc_cnt, seg_count);
		if (buffer == NULL)
			D_GOTO(out, rc = -DER_NOMEM);

		io->ic_csum_recalcs = buffer;
		io->ic_csum_recalc_cnt = seg_count;
	}

	iov.iov_buf_len = buf_max; /* for sanity check */
	i = lgc_seg->ls_idx_start;
	while (i <= lgc_seg->ls_idx_end) {
		if (lgc_seg->ls_phy_ent != NULL) {
			phy_ent = lgc_seg->ls_phy_ent;
			ext = ent_in->ei_rect.rc_ex;
		} else {
			struct agg_lgc_ent *lgc_ent = &mw->mw_lgc_ents[i];

			phy_ent = lgc_ent->le_phy_ent;
			ext = lgc_ent->le_ext;
		}
		i++;

		D_ASSERT(ext1_covers_ext2(&ent_in->ei_rect.rc_ex, &ext));
		D_ASSERT(ext1_covers_ext2(&phy_ent->pe_rect.rc_ex, &ext));

		phy_lo = phy_ent->pe_rect.rc_ex.ex_lo;
		if (phy_ent->pe_off != 0)
			phy_lo += phy_ent->pe_off;

		D_ASSERT(phy_lo <= phy_ent->pe_rect.rc_ex.ex_hi);
		D_ASSERT(ext.ex_lo >= phy_lo);

		copy_size = evt_extent_width(&ext) * ent_in->ei_inob;

		addr_src = phy_ent->pe_addr;
		addr_src.ba_off += (ext.ex_lo - phy_lo) * ent_in->ei_inob;

		D_ASSERT(!bio_addr_is_hole(&addr_src));
		D_ASSERT(iov.iov_buf_len >= copy_size);

		D_ASSERT(biov_idx < bsgl.bs_nr);
		bio_iov_set(&bsgl.bs_iovs[biov_idx], addr_src, copy_size);

		if (mw->mw_csum_support) {
			unsigned int wider = 0; /* length of per-ext csum add */
			/* Extends the biov entry to include additional
			 * data ranges (prefix, suffix) required for csum
			 * verification.
			 */
			unsigned int add_cnt =
				csum_widen_biov(&bsgl.bs_iovs[biov_idx],
						phy_ent, &ext,
						ent_in->ei_inob, phy_lo,
						&wider);

			/* add_cnt is the number of data segments to add. */
			if (add_cnt) {
				buf_add += wider;
				iov.iov_buf_len += wider;
				copy_size += wider;
				added_csum_segs += add_cnt;
			}
			csum_add_recalcs(&io->ic_csum_recalcs, phy_ent, &ext,
					 &bsgl, biov_idx);
		}
		biov_idx++;
		D_ASSERT(iov.iov_buf_len >= copy_size);
		iov.iov_buf_len -= copy_size;
	}

	D_ASSERT(seg_size == buf_max - iov.iov_buf_len);

	/* Moved read buf allocation to after loop, to allow inclusion
	 * of additional data needed for verification of prior checksums.
	 */
	if (io->ic_buf_len < buf_max + buf_add) {
		void *buffer;

		D_REALLOC(buffer, io->ic_buf, io->ic_buf_len,
			  buf_max + buf_add);
		if (buffer == NULL) {
			rc = -DER_NOMEM;
			goto out;
		}
		io->ic_buf = buffer;
		io->ic_buf_len = buf_max + buf_add;
	}

	if (added_csum_segs) {
		/* Additional data requird to verify checksums is read
		 * into end of read buffer. This allows the write data
		 * to be placed as a single contiguous range at beginning
		 * of the buffer.
		 */
		rc = csum_append_added_segs(&bsgl, added_csum_segs);
		if (rc) {
			D_ERROR("Extend bsgl error: "DF_RC"\n", DP_RC(rc));
			goto out;
		}
	}

	iov.iov_buf = io->ic_buf;
	iov.iov_len = 0;
	iov.iov_buf_len = io->ic_buf_len;
	sgl.sg_nr = 1;
	sgl.sg_iovs = &iov;
	rc = bio_readv(bio_ctxt, &bsgl, &sgl);
	if (rc) {
		D_ERROR("Readv for "DF_RECT" error: "DF_RC"\n",
			DP_RECT(&ent_in->ei_rect), DP_RC(rc));
		goto out;
	}
	D_ASSERT(iov.iov_len == seg_size + buf_add);

	if (mw->mw_csum_support) {
		/* Verify prior data, calculate csums for output range. */
		rc = csum_recalc(io, &bsgl, &sgl, ent_in, io->ic_csum_recalcs,
				 seg_count, seg_size);
		if (rc) {
			D_ERROR("CSUM verify error: "DF_RC"\n", DP_RC(rc));
			goto out;
		}
	}

	/* For csum support, this has moved reserve to after read, in case
	 * there's a csum mismatch on the verification of the read data.
	 * In case of a verification mismatch, the output extent is not
	 * reserved or inserted, and the data is not written to media.
	 */
	rc = reserve_segment(obj, io, seg_size, &ent_in->ei_addr);
	if (rc) {
		D_ERROR("Reserve "DF_U64" segment error: "DF_RC"\n", seg_size,
			DP_RC(rc));
		goto out;
	}

	addr_dst = ent_in->ei_addr;
	D_ASSERT(!bio_addr_is_hole(&addr_dst));

	iov.iov_buf = io->ic_buf;
	iov.iov_buf_len = io->ic_buf_len;
	iov.iov_len = seg_size;
	rc = bio_write(bio_ctxt, addr_dst, &iov);
	if (rc)
		D_ERROR("Write "DF_RECT" error: "DF_RC"\n",
			DP_RECT(&ent_in->ei_rect), DP_RC(rc));
out:
	bio_sgl_fini(&bsgl);
	return rc;
}

static int
fill_segments(daos_handle_t ih, struct agg_merge_window *mw,
	      unsigned int *acts)
{
	struct agg_io_context	*io = &mw->mw_io_ctxt;
	struct agg_lgc_seg	*lgc_seg;
	struct pobj_action	*scm_exts;
	unsigned int		 i, scm_max;
	int			 rc = 0;

	if (io->ic_seg_cnt == 0) {
		/** No logical extent or truncated physical extent (only removals) */
		return 0;
	}

	scm_max = MAX(io->ic_seg_cnt, 200);
	if (io->ic_rsrvd_scm == NULL ||
	    io->ic_rsrvd_scm->rs_actv_cnt < scm_max) {
		struct vos_rsrvd_scm	*rsrvd_scm;
		size_t			 size;

		size = sizeof(*io->ic_rsrvd_scm) *
			sizeof(*scm_exts) * scm_max;

		D_REALLOC_Z(rsrvd_scm, io->ic_rsrvd_scm, size);
		if (rsrvd_scm == NULL)
			return -DER_NOMEM;

		io->ic_rsrvd_scm = rsrvd_scm;
		io->ic_rsrvd_scm->rs_actv_cnt = scm_max;
	}
	D_ASSERT(io->ic_rsrvd_scm->rs_actv_at == 0);

	for (i = 0; i < io->ic_seg_cnt; i++) {
		lgc_seg = &io->ic_segs[i];

		D_DEBUG(DB_EPC, "Fill segment: %u-%u "DF_RECT"\n",
			lgc_seg->ls_idx_start, lgc_seg->ls_idx_end,
			DP_RECT(&lgc_seg->ls_ent_in.ei_rect));

		rc = fill_one_segment(ih, mw, lgc_seg, acts);
		if (rc) {
			D_ERROR("Fill seg %u-%u %p "DF_RECT" error: "DF_RC"\n",
				lgc_seg->ls_idx_start, lgc_seg->ls_idx_end,
				lgc_seg->ls_phy_ent,
				DP_RECT(&lgc_seg->ls_ent_in.ei_rect),
					DP_RC(rc));
			break;
		}
	}

	return rc;
}

static int
process_removals(struct agg_merge_window *mw, struct vos_obj_iter *oiter, d_list_t *head, bool last,
		 bool top)
{
	struct agg_rmv_ent	*rm_ent, *rm_tmp;
	struct evt_rect		 rect;
	int			 rc = 0;

	d_list_for_each_entry_safe(rm_ent, rm_tmp, head, re_link) {
		rect = rm_ent->re_rect;

		if (!last && (rm_ent->re_phy_count != 0 || rect.rc_ex.ex_hi > mw->mw_ext.ex_hi))
			continue;

		if (!rm_ent->re_aggregate) {
			D_ASSERT(d_list_empty(&rm_ent->re_contained));
			D_DEBUG(DB_EPC, "Removing physical removal record: "DF_RECT"\n",
				DP_RECT(&rm_ent->re_rect));
			rc = evt_delete(oiter->it_hdl, &rect, NULL);
			d_list_del(&rm_ent->re_phy_link);
		} else if (!d_list_empty(&rm_ent->re_contained)) {
			D_ASSERT(top);
			D_DEBUG(DB_EPC, "Removing logical removal record: "DF_RECT"\n",
				DP_RECT(&rm_ent->re_rect));
			rc = process_removals(mw, oiter, &rm_ent->re_contained, last, false);

			if (!d_list_empty(&rm_ent->re_contained))
				continue;
		}
		if (rc) {
			D_ERROR("Remove "DF_RECT" error: "DF_RC"\n",
				DP_RECT(&rect), DP_RC(rc));
			return rc;
		}

		d_list_del(&rm_ent->re_link);
		if (top) {
			D_ASSERT(mw->mw_rmv_cnt > 0);
			mw->mw_rmv_cnt--;
		}
		D_FREE(rm_ent);
	}

	return 0;
}

static void
unmark_removals(struct agg_merge_window *mw, const struct agg_phy_ent *phy_ent)
{
	struct agg_rmv_ent	*rmv_ent;

	d_list_for_each_entry_reverse(rmv_ent, &mw->mw_phy_rmv_ents, re_phy_link) {
		if (rmv_ent->re_rect.rc_epc != phy_ent->pe_rect.rc_epc)
			continue;

		if (rmv_ent->re_rect.rc_ex.ex_hi < phy_ent->pe_rect.rc_ex.ex_lo)
			break;

		if (rmv_ent->re_rect.rc_ex.ex_lo > phy_ent->pe_rect.rc_ex.ex_hi)
			continue;

		D_ASSERT(rmv_ent->re_phy_count > 0);
		rmv_ent->re_phy_count--;
	}
}

static int
insert_segments(daos_handle_t ih, struct agg_merge_window *mw,
		bool last, unsigned int *acts)
{
	struct vos_obj_iter	*oiter = vos_hdl2oiter(ih);
	struct vos_object	*obj = oiter->it_obj;
	struct agg_io_context	*io = &mw->mw_io_ctxt;
	struct agg_phy_ent	*phy_ent, *tmp;
	struct agg_lgc_ent	*lgc_ent;
	struct agg_lgc_seg	*lgc_seg;
	struct evt_entry_in	*ent_in;
	struct evt_rect		 rect;
	unsigned int		 i, leftovers = 0;
	int			 rc;

	D_ASSERT(obj != NULL);
	rc = umem_tx_begin(vos_obj2umm(obj), NULL);
	if (rc)
		return rc;

	/* Publish SCM reservations */
	rc = vos_publish_scm(obj->obj_cont, io->ic_rsrvd_scm, true);
	if (rc) {
		D_ERROR("Publish SCM extents error: "DF_RC"\n", DP_RC(rc));
		goto abort;
	}

	/* Adjust logical entry queue */
	for (i = 0; i < mw->mw_lgc_cnt; i++) {
		lgc_ent = &mw->mw_lgc_ents[i];
		phy_ent = lgc_ent->le_phy_ent;

		D_ASSERT(ext1_covers_ext2(&mw->mw_ext, &lgc_ent->le_ext));
		D_ASSERT(phy_ent->pe_ref > 0);
		phy_ent->pe_ref--;
		phy_ent->pe_trunc_head = true;
	}
	mw->mw_lgc_cnt = 0;

	/* Adjust payload address of truncated physical entries */
	for (i = 0; i < io->ic_seg_cnt; i++) {
		lgc_seg = &io->ic_segs[i];
		ent_in = &io->ic_segs[i].ls_ent_in;
		phy_ent = lgc_seg->ls_phy_ent;

		if (phy_ent != NULL && !bio_addr_is_hole(&ent_in->ei_addr)) {
			phy_ent->pe_addr = ent_in->ei_addr;
			/* Checksum from ent_in is assigned to truncated
			 * physical entry, in addition to re-assigning address.
			 */
			phy_ent->pe_csum_info = ent_in->ei_csum;
		}
	}

	/* Remove old physical entries from EV tree */
	d_list_for_each_entry_safe(phy_ent, tmp, &mw->mw_phy_ents, pe_link) {
		rect = phy_ent->pe_rect;

		D_ASSERT(phy_ent->pe_ref == 0);
		/* The physical entry was truncated on prev window flush */
		if (phy_ent->pe_off != 0)
			rect.rc_ex.ex_lo += phy_ent->pe_off;

		D_ASSERTF(rect.rc_ex.ex_lo <= rect.rc_ex.ex_hi, "phy_ent "DF_RECT" off="DF_X64"\n",
			  DP_RECT(&phy_ent->pe_rect), phy_ent->pe_off);
		D_ASSERT(phy_ent->pe_remove || rect.rc_ex.ex_lo <= mw->mw_ext.ex_hi);

		/*
		 * The physical entry spans window end, but is fully covered
		 * in current window, keep it intact.
		 */
		if (!phy_ent->pe_remove && rect.rc_ex.ex_hi > mw->mw_ext.ex_hi &&
		    !phy_ent->pe_trunc_head) {
			leftovers++;
			continue;
		}

		rc = evt_delete(oiter->it_hdl, &rect, NULL);
		if (rc) {
			D_ERROR("Delete "DF_RECT" pe_off:"
				DF_U64" error: "DF_RC"\n",
				DP_RECT(&rect), phy_ent->pe_off,
				DP_RC(rc));
			goto abort;
		}

		/* Physical entry is in window or fully removed */
		if (rect.rc_ex.ex_hi <= mw->mw_ext.ex_hi ||
		    phy_ent->pe_remove) {
			d_list_del(&phy_ent->pe_link);
			unmark_removals(mw, phy_ent);
			D_FREE(phy_ent);
			D_ASSERT(mw->mw_phy_cnt > 0);
			mw->mw_phy_cnt--;
			continue;
		}

		/* Update extent start of truncated physical entry */
		rect.rc_ex.ex_lo = mw->mw_ext.ex_hi + 1;
		phy_ent->pe_off = rect.rc_ex.ex_lo -
				phy_ent->pe_rect.rc_ex.ex_lo;
		phy_ent->pe_trunc_head = false;

		leftovers++;
	}
	D_ASSERT(leftovers == mw->mw_phy_cnt);

	/** Remove processed removal records */
	rc = process_removals(mw, oiter, &mw->mw_rmv_ents, last, true);

	/* Insert new segments into EV tree */
	for (i = 0; i < io->ic_seg_cnt; i++) {
		ent_in = &io->ic_segs[i].ls_ent_in;

		/** For insertion, no tx will be inserting anything at this
		 *  epoch so just use the max value for the minor epoch.
		 */
		rc = evt_insert(oiter->it_hdl, ent_in,
				&ent_in->ei_csum.cs_csum);
		if (rc) {
			D_ERROR("Insert segment "DF_RECT" error: "DF_RC"\n",
				DP_RECT(&ent_in->ei_rect), DP_RC(rc));
			goto abort;
		}
	}

	/* Clear window size */
	mw->mw_ext.ex_lo = mw->mw_ext.ex_hi = mw->mw_alloc_hi = 0;

	/* Publish NVMe reservations */
	rc = vos_publish_blocks(obj->obj_cont, &io->ic_nvme_exts, true,
				VOS_IOS_AGGREGATION);
	if (rc) {
		D_ERROR("Publish NVMe extents error: "DF_RC"\n", DP_RC(rc));
		goto abort;
	}
abort:
	if (rc)
		rc = umem_tx_abort(vos_obj2umm(obj), rc);
	else
		rc = umem_tx_commit(vos_obj2umm(obj));

	return rc;
}

static void
cleanup_segments(daos_handle_t ih, struct agg_merge_window *mw, int rc)
{
	struct vos_obj_iter	*oiter = vos_hdl2oiter(ih);
	struct vos_object	*obj = oiter->it_obj;
	struct agg_io_context	*io = &mw->mw_io_ctxt;

	D_ASSERT(obj != NULL);
	if (rc) {
		vos_publish_scm(obj->obj_cont, io->ic_rsrvd_scm, false);

		if (!d_list_empty(&io->ic_nvme_exts))
			vos_publish_blocks(obj->obj_cont, &io->ic_nvme_exts,
					   false, VOS_IOS_AGGREGATION);
	}

	/* Reset io context */
	D_ASSERT(d_list_empty(&io->ic_nvme_exts));
	D_ASSERT(io->ic_rsrvd_scm == NULL ||
		 io->ic_rsrvd_scm->rs_actv_at == 0);
	io->ic_seg_cnt = 0;
}

static void
clear_merge_window(struct agg_merge_window *mw)
{
	struct agg_phy_ent *phy_ent, *tmp;

	mw->mw_ext.ex_lo = mw->mw_ext.ex_hi = mw->mw_alloc_hi = 0;
	mw->mw_lgc_cnt = 0;
	d_list_for_each_entry_safe(phy_ent, tmp, &mw->mw_phy_ents, pe_link) {
		d_list_del(&phy_ent->pe_link);
		unmark_removals(mw, phy_ent);
		D_FREE(phy_ent);
	}
	mw->mw_phy_cnt = 0;
}

static void
free_removal_records(struct agg_merge_window *mw, d_list_t *head, bool top)
{
	struct agg_rmv_ent *rm_ent, *tmp;

	d_list_for_each_entry_safe(rm_ent, tmp, head, re_link) {
		d_list_del(&rm_ent->re_link);
		if (!d_list_empty(&rm_ent->re_contained)) {
			D_ASSERT(top);
			free_removal_records(mw, &rm_ent->re_contained, false);
		}
		D_FREE(rm_ent);
	}
	if (top) {
		mw->mw_rmv_cnt = 0;
		D_INIT_LIST_HEAD(&mw->mw_phy_rmv_ents);
	}
}

static bool
need_flush(struct agg_merge_window *mw, bool last)
{
	struct agg_phy_ent	*phy_ent;
	struct agg_lgc_ent	*lgc_ent;
	struct evt_extent	 lgc_ext, phy_ext;
	int			 i;
	bool			 hole = false;

	for (i = 0; i < mw->mw_lgc_cnt; i++) {
		lgc_ent = &mw->mw_lgc_ents[i];
		phy_ent = lgc_ent->le_phy_ent;
		phy_ext = phy_ent->pe_rect.rc_ex;
		lgc_ext = lgc_ent->le_ext;

		/*
		 * Any physical entry is partially covered, or appeared
		 * in other window
		 */
		if (lgc_ext.ex_lo != phy_ext.ex_lo ||
		    lgc_ext.ex_hi != phy_ext.ex_hi)
			return true;

		/* If any consecutive visible entries can be merged */
		if (i != 0 && hole == bio_addr_is_hole(&phy_ent->pe_addr))
			return true;

		hole = bio_addr_is_hole(&phy_ent->pe_addr);
	}

	/* Any invisible physical entries ? */
	if (mw->mw_lgc_cnt != mw->mw_phy_cnt)
		return true;

	/* Need to cleanup remaining removal records */
	if (last && mw->mw_rmv_cnt != 0)
		return true;

	clear_merge_window(mw);
	D_DEBUG(DB_EPC, "Skip window flush "DF_EXT"\n", DP_EXT(&mw->mw_ext));

	return false;
}

static int
flush_merge_window(daos_handle_t ih, struct agg_merge_window *mw,
		   bool last, unsigned int *acts)
{
	int	rc;

	/*
	 * If no new updates in an already aggregated window, window flush will
	 * be skipped, otherwise, all the data within the window will be
	 * migrated to a new location, such batch data migration is good for
	 * anti-fragmentaion.
	 */
	if (!need_flush(mw, last))
		return 0;

	/* Prepare the new segments to be inserted */
	rc = prepare_segments(mw);
	if (rc) {
		D_ERROR("Prepare segments "DF_EXT" error: "DF_RC"\n",
			DP_EXT(&mw->mw_ext), DP_RC(rc));
		goto out;
	}

	/* Transfer data from old logical records to reserved new segments */
	rc = fill_segments(ih, mw, acts);
	if (rc) {
		D_ERROR("Fill segments "DF_EXT" error: "DF_RC"\n",
			DP_EXT(&mw->mw_ext), DP_RC(rc));
		goto out;
	}

	/* Replace the old logical records with new segments in EV tree */
	rc = insert_segments(ih, mw, last, acts);
	if (rc) {
		D_ERROR("Insert segments "DF_EXT" error: "DF_RC"\n",
			DP_EXT(&mw->mw_ext), DP_RC(rc));
		goto out;
	}
out:
	cleanup_segments(ih, mw, rc);
	return rc;
}

static bool
trigger_flush(struct agg_merge_window *mw, struct evt_extent *lgc_ext)
{
	struct evt_extent *w_ext = &mw->mw_ext;

	D_ASSERT(w_ext->ex_lo <= lgc_ext->ex_lo);
	/* Empty or closed merge window */
	if (merge_window_status(mw) == MW_CLOSED ||
	    merge_window_status(mw) == MW_FLUSHED)
		return false;

	/*
	 * Window is formed by visible logical entries, must have no
	 * overlapping.
	 */
	D_ASSERTF(w_ext->ex_hi < lgc_ext->ex_lo, "win:"DF_EXT", lgc_ent:"
		  DF_EXT"\n", DP_EXT(w_ext), DP_EXT(lgc_ext));

	/* Window is large enough */
	if (merge_window_size(mw) >= mw->mw_flush_thresh)
		return true;

	/* Trigger flush when entry is disjoint with window */
	return !((w_ext->ex_hi + 1) == lgc_ext->ex_lo);
}

static struct agg_phy_ent *
enqueue_phy_ent(struct agg_merge_window *mw, struct evt_extent *phy_ext,
		const vos_iter_entry_t *entry, bio_addr_t *addr,
		struct dcs_csum_info *csum_info, uint32_t ver)
{
	struct agg_phy_ent *phy_ent;

	D_ALLOC_PTR(phy_ent);
	if (phy_ent == NULL)
		return NULL;

	phy_ent->pe_rect.rc_ex = *phy_ext;
	phy_ent->pe_rect.rc_epc = entry->ie_epoch;
	phy_ent->pe_rect.rc_minor_epc = entry->ie_minor_epc;
	phy_ent->pe_addr = *addr;
	phy_ent->pe_csum_info = *csum_info;
	phy_ent->pe_off = 0;
	phy_ent->pe_ver = ver;
	phy_ent->pe_ref = 0;

	/* Physical entry with valid csum type triggers checksum recalcuation.
	 */
	if (phy_ent->pe_csum_info.cs_type && mw->mw_csum_support == false)
		mw->mw_csum_support = true;
	else if (!phy_ent->pe_csum_info.cs_type && mw->mw_csum_support == true)
		mw->mw_csum_support = false;

	/* Sanity check */
	if (!d_list_empty(&mw->mw_phy_ents)) {
		struct agg_phy_ent *prev;

		D_ASSERT(mw->mw_phy_cnt != 0);
		prev = d_list_entry(mw->mw_phy_ents.prev, struct agg_phy_ent,
				    pe_link);
		D_ASSERTF(prev->pe_rect.rc_ex.ex_lo <= phy_ext->ex_lo,
			  "prev phy_ext: "DF_EXT", phy_ext: "DF_EXT"\n",
			  DP_EXT(&prev->pe_rect.rc_ex), DP_EXT(phy_ext));
	} else {
		D_ASSERT(mw->mw_phy_cnt == 0);
	}

	d_list_add_tail(&phy_ent->pe_link, &mw->mw_phy_ents);
	mw->mw_phy_cnt++;

	return phy_ent;
}

static int
enqueue_lgc_ent(struct agg_merge_window *mw, struct evt_extent *lgc_ext,
		struct agg_phy_ent *phy_ent)
{
	struct agg_lgc_ent	*lgc_ent;
	unsigned int		 max, cnt;

	max = mw->mw_lgc_max;
	cnt = mw->mw_lgc_cnt;
	/* Sanity check */
	if (cnt > 0) {
		lgc_ent = &mw->mw_lgc_ents[cnt - 1];
		D_ASSERTF(lgc_ext->ex_lo == lgc_ent->le_ext.ex_hi + 1 &&
			  lgc_ent->le_ext.ex_hi == mw->mw_ext.ex_hi,
			  "prev lgc_ext: "DF_EXT", lgc_ext: "DF_EXT"\n",
			  DP_EXT(&lgc_ent->le_ext), DP_EXT(lgc_ext));
	}

	if (cnt == max) {
		unsigned int new_max = max ? max * 2 : 10;

		D_REALLOC_ARRAY(lgc_ent, mw->mw_lgc_ents, max, new_max);
		if (lgc_ent == NULL)
			return -DER_NOMEM;

		mw->mw_lgc_max = new_max;
		mw->mw_lgc_ents = lgc_ent;
	}

	D_ASSERT(mw->mw_lgc_max > mw->mw_lgc_cnt);
	lgc_ent = &mw->mw_lgc_ents[cnt];
	lgc_ent->le_ext = *lgc_ext;
	phy_ent->pe_ref++;
	lgc_ent->le_phy_ent = phy_ent;
	mw->mw_lgc_cnt++;

	/*
	 * Extend window size. If the visible entry is a punched record, the
	 * window size could be very huge, but this is ok, because there won't
	 * be huge contiguous allocation on window flush, only lots of covered
	 * physical entries being deleted.
	 */
	if (mw->mw_lgc_cnt == 1)
		mw->mw_ext.ex_lo = lgc_ext->ex_lo;
	mw->mw_ext.ex_hi = mw->mw_alloc_hi = lgc_ext->ex_hi;

	D_DEBUG(DB_EPC, "lgc_ext:"DF_EXT", phy_ext:"DF_RECT", mw:"DF_EXT", "
		"index:%u\n", DP_EXT(lgc_ext), DP_RECT(&phy_ent->pe_rect),
		DP_EXT(&mw->mw_ext), cnt);

	return 0;
}

static void
close_merge_window(struct agg_merge_window *mw, int rc)
{
	struct agg_io_context *io = &mw->mw_io_ctxt;

	if (rc) {
		clear_merge_window(mw);
		free_removal_records(mw, &mw->mw_rmv_ents, true);
	}

	D_ASSERT(mw->mw_rmv_cnt == 0);
	D_ASSERT(merge_window_status(mw) != MW_OPENED);

	mw->mw_rsize = 0;
	if (mw->mw_lgc_ents != NULL) {
		D_FREE(mw->mw_lgc_ents);
		mw->mw_lgc_ents = NULL;
		mw->mw_lgc_max = 0;
	}

	if (io->ic_buf != NULL) {
		D_FREE(io->ic_buf);
		io->ic_buf = NULL;
		io->ic_buf_len = 0;
	}

	if (io->ic_segs != NULL) {
		D_FREE(io->ic_segs);
		io->ic_segs = NULL;
		io->ic_seg_max = 0;
	}

	D_FREE(io->ic_rsrvd_scm);

	if (io->ic_csum_recalcs != NULL) {
		D_FREE(io->ic_csum_recalcs);
		io->ic_csum_recalcs = NULL;
		io->ic_csum_recalc_cnt = 0;
	}
	if (io->ic_csum_buf != NULL) {
		D_FREE(io->ic_csum_buf);
		io->ic_csum_buf = NULL;
		io->ic_csum_buf_len = 0;
	}
}

static struct agg_phy_ent *
lookup_phy_ent(struct agg_merge_window *mw, const struct evt_extent *phy_ext,
	       const vos_iter_entry_t *entry)
{
	struct agg_phy_ent *phy_ent;

	d_list_for_each_entry_reverse(phy_ent, &mw->mw_phy_ents, pe_link) {
		/* Physical entry list is sorted by extent start */
		if (phy_ent->pe_rect.rc_ex.ex_lo < phy_ext->ex_lo)
			break;

		if (phy_ent->pe_rect.rc_epc == entry->ie_epoch &&
		    phy_ent->pe_rect.rc_minor_epc == entry->ie_minor_epc &&
		    phy_ent->pe_rect.rc_ex.ex_hi == phy_ext->ex_hi)
			return phy_ent;
	}

	return NULL;
}

static void
mark_removals(struct agg_merge_window *mw, struct agg_phy_ent *phy_ent,
	      const struct evt_extent *lgc_ext)
{
	struct agg_rmv_ent	*rmv_ent;

	if (d_list_empty(&mw->mw_phy_rmv_ents))
		return;

	/* This is not a real entry but it doesn't matter.   It will be used to calculate
	 * where to continue if the physical record has been processed before to ensure
	 * we never refcount the same record more than once.
	 */
	rmv_ent = d_list_entry(&mw->mw_phy_rmv_ents, struct agg_rmv_ent, re_phy_link);

	d_list_for_each_entry_continue(rmv_ent, &mw->mw_phy_rmv_ents, re_phy_link) {
		if (rmv_ent->re_rect.rc_epc != phy_ent->pe_rect.rc_epc)
			continue;

		if (rmv_ent->re_rect.rc_ex.ex_hi < lgc_ext->ex_lo)
			continue;

		/** We should be processing extents in order so this should mean there is overlap */
		D_ASSERT(rmv_ent->re_rect.rc_ex.ex_lo <= lgc_ext->ex_lo);
		rmv_ent->re_phy_count++;
	}
}

static int
join_merge_window(daos_handle_t ih, struct agg_merge_window *mw,
		  vos_iter_entry_t *entry, unsigned int *acts)
{
	struct vos_obj_iter	*oiter = vos_hdl2oiter(ih);
	struct evt_extent	 phy_ext, lgc_ext;
	struct agg_phy_ent	*phy_ent;
	bool			 remove, visible, partial, last;
	int			 rc = 0;

	recx2ext(&entry->ie_recx, &lgc_ext);
	recx2ext(&entry->ie_orig_recx, &phy_ext);
	D_ASSERT(ext1_covers_ext2(&phy_ext, &lgc_ext));

	switch (entry->ie_dtx_state) {
	case DTX_ST_COMMITTED:
		break;
	case DTX_ST_ABORTED:
		/*
		 * Delete the aborted entry, and inform iterator to abort
		 * current evtree aggregation.
		 *
		 * NB. We can't continue current evtree aggregation since
		 * other entry's visibility could be invalid after deleting.
		 */
		D_DEBUG(DB_EPC, "Delete aborted EV entry "DF_EXT"@"DF_X64"\n",
			DP_EXT(&phy_ext), entry->ie_epoch);

		rc = delete_evt_entry(oiter, entry, "aborted");
		if (rc)
			return rc;
		/** We just need an alternative error code.  Use -DER_TX_RESTART
		 *  here to indicate that we hit an aborted entry and need to
		 *  restart the aggregation of the evtree.  Using -DER_TX_BUSY
		 *  would mean aborting the current level and everything above
		 *  it.   We only want to do that if we hit an in-progress
		 *  entry.
		 */
		return -DER_TX_RESTART;
	case DTX_ST_PREPARED:
		/*
		 * Keep uncommitted entry, and inform iterator to abort
		 * current evtree aggregation.
		 */
		D_DEBUG(DB_EPC, "Hit uncommitted EV entry "DF_EXT"@"DF_X64"\n",
			DP_EXT(&phy_ext), entry->ie_epoch);
		return -DER_TX_BUSY;
	default:
		D_ASSERTF(0, "Unexpected DTX state: %d\n", entry->ie_dtx_state);
		break;
	}

	visible = (entry->ie_vis_flags & VOS_VIS_FLAG_VISIBLE);
	remove = (entry->ie_vis_flags & VOS_VIS_FLAG_REMOVE);
	partial = (entry->ie_vis_flags & VOS_VIS_FLAG_PARTIAL);
	last = (entry->ie_vis_flags & VOS_VIS_FLAG_LAST);

	/* Just delete the fully covered intact physical entry */
	if (!visible && !partial && !remove) {
		D_ASSERTF(lgc_ext.ex_lo == phy_ext.ex_lo &&
			  lgc_ext.ex_hi == phy_ext.ex_hi,
			  ""DF_EXT" != "DF_EXT"\n",
			  DP_EXT(&lgc_ext), DP_EXT(&phy_ext));
		D_ASSERT(entry->ie_vis_flags & VOS_VIS_FLAG_COVERED);

		rc = delete_evt_entry(oiter, entry, "covered");
		if (rc)
			return rc;
		goto out;
	}

	if (remove) {
		/* Enqueue removal record */
		rc = enqueue_rmv_ent(mw, &phy_ext, entry, oiter);
		if (rc != 0) {
			D_ERROR("Enqueue rm_ent win:"DF_EXT", ent:"DF_EXT" "
				"error: "DF_RC"\n", DP_EXT(&mw->mw_ext),
				DP_EXT(&phy_ext), DP_RC(rc));
			return rc;
		}

		goto out;
	}

	/* Trigger current window flush when reaching threshold */
	if (visible && trigger_flush(mw, &lgc_ext)) {
		/* The window flush doesn't expect holes caused by removal records */
		mw->mw_ext.ex_hi = lgc_ext.ex_lo - 1;
		rc = flush_merge_window(ih, mw, false, acts);
		if (rc) {
			D_ERROR("Flush window "DF_EXT" error: "DF_RC"\n",
				DP_EXT(&mw->mw_ext), DP_RC(rc));
			return rc;
		}
		D_ASSERT(merge_window_status(mw) == MW_FLUSHED);
	}

	/* Lookup physical entry, enqueue if it doesn't exist */
	phy_ent = lookup_phy_ent(mw, &phy_ext, entry);
	if (phy_ent == NULL) {
		if (phy_ext.ex_lo != lgc_ext.ex_lo) {
			D_ASSERT(!visible && phy_ent_is_removed(mw, &phy_ext, entry->ie_epoch));
			goto out;
		}
		phy_ent = enqueue_phy_ent(mw, &phy_ext, entry,
					  &entry->ie_biov.bi_addr,
					  &entry->ie_csum, entry->ie_ver);
		if (phy_ent == NULL) {
			rc = -DER_NOMEM;
			D_ERROR("Enqueue phy_ent win:"DF_EXT", ent:"DF_EXT" "
				"error: "DF_RC"\n", DP_EXT(&mw->mw_ext),
				DP_EXT(&phy_ext), DP_RC(rc));
			return rc;
		}
	} else {
		/* Can't be the first logcial entry */
		D_ASSERT(phy_ext.ex_lo != lgc_ext.ex_lo);
	}

	/* Enqueue the visible logical entry */
	if (visible) {
		rc = enqueue_lgc_ent(mw, &lgc_ext, phy_ent);
		if (rc) {
			D_ERROR("Enqueue lgc_ent win: "DF_EXT", ent:"DF_EXT" "
				"error: "DF_RC"\n", DP_EXT(&mw->mw_ext),
				DP_EXT(&lgc_ext), DP_RC(rc));
			return rc;
		}
	} else {
		/* Fully covered physical entry must have been deleted */
		D_ASSERT(partial);
		/* refcount any removal records covering this extent */
		mark_removals(mw, phy_ent, &lgc_ext);
	}
out:
	/* Flush & close window on last entry */
	if (last) {
		rc = flush_merge_window(ih, mw, true, acts);
		if (rc)
			D_ERROR("Flush window "DF_EXT" error: "DF_RC"\n",
				DP_EXT(&mw->mw_ext), DP_RC(rc));

		close_merge_window(mw, rc);
	}

	return rc;
}

static int
set_window_size(struct agg_merge_window *mw, daos_size_t rsize)
{
	int	rc = 0;

	if (rsize == 0) {
		D_DEBUG(DB_TRACE, "EV tree 0 iod_size could be caused by "
			"inserting punch records in an empty tree.  This can "
			"happen during rebuild.\n");
		/** Just set it to 1.  If the tree is all holes anyway, it
		 *  should be fine to assume a record size.
		 */
		rsize = 1;
	}

	if (mw->mw_rsize == 0) {
		mw->mw_rsize = rsize;

		if (DAOS_FAIL_CHECK(DAOS_VOS_AGG_MW_THRESH)) {
			mw->mw_flush_thresh = daos_fail_value_get();
			D_INFO("Set flush threshold to: "DF_U64"\n",
			       mw->mw_flush_thresh);
		} else if (rsize < (VOS_MW_FLUSH_THRESH / 2)) {
			mw->mw_flush_thresh = VOS_MW_FLUSH_THRESH;
		} else {
			mw->mw_flush_thresh = (rsize < VOS_MW_FLUSH_THRESH) ?
						rsize * 2 : rsize;
			D_INFO("Bump flush threshold to: "DF_U64", rsize: "
			       ""DF_U64"\n", mw->mw_flush_thresh, rsize);
		}
	} else if (mw->mw_rsize != rsize) {
		D_CRIT("Mismatched iod_size "DF_U64" != "DF_U64"\n",
		       mw->mw_rsize, rsize);
		rc = -DER_INVAL;
	}

	return rc;
}

static int
vos_agg_ev(daos_handle_t ih, vos_iter_entry_t *entry,
	   struct vos_agg_param *agg_param, unsigned int *acts)
{
	struct agg_merge_window	*mw = &agg_param->ap_window;
	struct evt_extent	 phy_ext, lgc_ext;
	int			 rc = 0;

	D_ASSERT(agg_param != NULL);
	D_ASSERT(acts != NULL);
	recx2ext(&entry->ie_recx, &lgc_ext);
	recx2ext(&entry->ie_orig_recx, &phy_ext);

	/* Discard */
	if (agg_param->ap_discard) {
		struct vos_obj_iter	*oiter = vos_hdl2oiter(ih);

		/*
		 * Delete the physical entry when iterating to the first
		 * logical entry
		 */
		if (phy_ext.ex_lo == lgc_ext.ex_lo)
			rc = delete_evt_entry(oiter, entry, "discarded");

		/*
		 * Sorted iteration doesn't support tree empty check, so we
		 * always inform vos_iterate() to check if subtree is empty.
		 */
		if (entry->ie_vis_flags & VOS_VIS_FLAG_LAST) {
			/* Trigger re-probe in akey iteration */
			*acts |= VOS_ITER_CB_YIELD;
		}
		return rc;
	}

	/* Aggregation Yield for testing purpose */
	while (DAOS_FAIL_CHECK(DAOS_VOS_AGG_BLOCKED))
		ABT_thread_yield();

	/* Aggregation */
	D_DEBUG(DB_EPC, "oid:"DF_UOID", lgc_ext:"DF_EXT", "
		"phy_ext:"DF_EXT", epoch:"DF_X64".%d, flags: %x(%c)\n",
		DP_UOID(agg_param->ap_oid), DP_EXT(&lgc_ext),
		DP_EXT(&phy_ext), entry->ie_epoch, entry->ie_minor_epc,
		entry->ie_vis_flags, evt_vis2dbg(entry->ie_vis_flags));

	rc = set_window_size(mw, entry->ie_rsize);
	if (rc)
		goto out;

	rc = join_merge_window(ih, mw, entry, acts);
	if (rc)
		D_CDEBUG(rc == -DER_TX_RESTART || rc == -DER_TX_BUSY, DB_TRACE,
			 DLOG_ERR, "Join window "DF_EXT"/"DF_EXT" error: "
			 DF_RC"\n", DP_EXT(&mw->mw_ext), DP_EXT(&phy_ext),
			 DP_RC(rc));
out:
	if (rc)
		close_merge_window(mw, rc);

	return rc;
}

static inline bool
vos_aggregate_yield(struct vos_agg_param *agg_param)
{
	if (agg_param->ap_yield_func != NULL)
		return agg_param->ap_yield_func(agg_param->ap_yield_arg);

	bio_yield();
	return false;
}

static int
vos_aggregate_pre_cb(daos_handle_t ih, vos_iter_entry_t *entry,
		     vos_iter_type_t type, vos_iter_param_t *param,
		     void *cb_arg, unsigned int *acts)
{
	struct vos_agg_param	*agg_param = cb_arg;
	struct vos_container	*cont;
	int			 rc = 0;

	cont = vos_hdl2cont(param->ip_hdl);
	D_DEBUG(DB_EPC, DF_CONT": Aggregate pre, type:%d, is_discard:%d\n",
		DP_CONT(cont->vc_pool->vp_id, cont->vc_id), type,
		agg_param->ap_discard);

	switch (type) {
	case VOS_ITER_OBJ:
		rc = vos_agg_obj(ih, entry, agg_param, acts);
		break;
	case VOS_ITER_DKEY:
		rc = vos_agg_dkey(ih, entry, agg_param, acts);
		break;
	case VOS_ITER_AKEY:
		rc = vos_agg_akey(ih, entry, agg_param, acts);
		break;
	case VOS_ITER_RECX:
		rc = vos_agg_ev(ih, entry, agg_param, acts);
		if (rc == -DER_TX_RESTART) {
			D_DEBUG(DB_EPC, "Restarting evtree aggregation\n");
			*acts |= VOS_ITER_CB_RESTART;
			rc = 0;
			break;
		}
		/* fall through to check for abort */
	case VOS_ITER_SINGLE:
		if (type == VOS_ITER_SINGLE)
			rc = vos_agg_sv(ih, entry, agg_param, acts);
		if (rc == -DER_CSUM || rc == -DER_TX_BUSY) {
			D_DEBUG(DB_EPC, "Abort value aggregation "DF_RC"\n",
				DP_RC(rc));

			*acts |= VOS_ITER_CB_ABORT;
			if (rc == -DER_CSUM) {
				agg_param->ap_csum_err = true;
			} else if (rc == -DER_TX_BUSY) {
				/** Must not aggregate anything above
				 *  this entry to avoid orphaned tree
				 *  assertion
				 */
				agg_param->ap_skip_akey = true;
				agg_param->ap_skip_dkey = true;
				agg_param->ap_skip_obj = true;
			}
			rc = 0;
		}
		break;
	default:
		D_ASSERTF(false, "Invalid iter type\n");
		rc = -DER_INVAL;
		break;
	}

	if (rc < 0) {
		D_ERROR("VOS aggregation failed: "DF_RC"\n", DP_RC(rc));
		return rc;
	}

	agg_param->ap_credits++;

	if (agg_param->ap_credits > agg_param->ap_credits_max ||
	    (DAOS_FAIL_CHECK(DAOS_VOS_AGG_RANDOM_YIELD) && (rand() % 2))) {
		D_DEBUG(DB_EPC, "Credits exhausted, type:%u, acts:%u\n",
			type, *acts);

		agg_param->ap_credits = 0;

		/*
		 * Reset position if we yield while iterating in object, dkey
		 * or akey level, so that subtree won't be skipped mistakenly,
		 * see the comment in vos_agg_obj().
		 *
		 * If current object/dkey/akey has been marked as processed,
		 * don't reset the position, otherwise, iterator will reprobe
		 * the same item and process it again.
		 */
		if (!(*acts & VOS_ITER_CB_SKIP))
			reset_agg_pos(type, agg_param);

		if (vos_aggregate_yield(agg_param)) {
			D_DEBUG(DB_EPC, "VOS discard/aggregation aborted\n");
			return 1;
		}
	}

	return 0;
}

static int
vos_aggregate_post_cb(daos_handle_t ih, vos_iter_entry_t *entry,
		      vos_iter_type_t type, vos_iter_param_t *param,
		      void *cb_arg, unsigned int *acts)
{
	struct vos_agg_param	*agg_param = cb_arg;
	struct vos_container	*cont;
	int			 rc = 0;

	cont = vos_hdl2cont(param->ip_hdl);
	D_DEBUG(DB_EPC, DF_CONT": Aggregate post, type:%d, is_discard:%d\n",
		DP_CONT(cont->vc_pool->vp_id, cont->vc_id), type,
		agg_param->ap_discard);

	switch (type) {
	case VOS_ITER_OBJ:
		if (agg_param->ap_skip_obj) {
			agg_param->ap_skip_obj = false;
			break;
		}
		rc = oi_iter_aggregate(ih, agg_param->ap_discard_obj);
		break;
	case VOS_ITER_DKEY:
		if (agg_param->ap_skip_dkey) {
			agg_param->ap_skip_dkey = false;
			break;
		}
	case VOS_ITER_AKEY:
		if (agg_param->ap_skip_akey) {
			agg_param->ap_skip_akey = false;
			break;
		}
		rc = vos_obj_iter_aggregate(ih, agg_param->ap_discard_obj);
		break;
	case VOS_ITER_SINGLE:
		return 0;
	case VOS_ITER_RECX:
		return 0;
	default:
		D_ASSERTF(false, "Invalid iter type\n");
		return -DER_INVAL;
	}

	if (rc == 1) {
		/* Reprobe flag is set */
		*acts |= VOS_ITER_CB_DELETE;
		rc = 0;
	} else if (rc != 0) {
		D_ERROR("VOS aggregation failed: %d\n", rc);

		/*
		 * -DER_TX_BUSY error indicates current ilog aggregation
		 * aborted on hitting uncommitted entry, this should be a very
		 * rare case, we'd suppress the error here to keep aggregation
		 * moving forward.   We do, however, need to ensure we do not
		 * aggregate anything in the parent path.  Otherwise, we could
		 * orphan the current entry due to incarnation log semantics.
		 */
		if (rc == -DER_TX_BUSY) {
			rc = 0;
			switch (type) {
			default:
				D_ASSERTF(type == VOS_ITER_OBJ,
					  "Invalid iter type\n");
				break;
			case VOS_ITER_AKEY:
				agg_param->ap_skip_dkey = true;
				/* fall through */
			case VOS_ITER_DKEY:
				agg_param->ap_skip_obj = true;
				/* fall through */
			}
		}
	}

	return rc;
}

enum {
	AGG_MODE_AGGREGATE,
	AGG_MODE_DISCARD,
	AGG_MODE_OBJ_DISCARD,
};

static int
aggregate_enter(struct vos_container *cont, int agg_mode,
		daos_epoch_range_t *epr)
{
	switch (agg_mode) {
	default:
		D_ASSERT(0);
		break;
	case AGG_MODE_DISCARD:
		if (cont->vc_in_discard) {
			D_ERROR(DF_CONT": Already in discard epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_discard.epr_lo, cont->vc_epr_discard.epr_hi);
			return -DER_BUSY;
		}

		if (cont->vc_obj_discard_count != 0) {
			D_ERROR(DF_CONT": In object discard epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_discard.epr_lo, cont->vc_epr_discard.epr_hi);
			return -DER_BUSY;
		}

		if (cont->vc_in_aggregation && cont->vc_epr_aggregation.epr_hi >= epr->epr_lo) {
			D_ERROR(DF_CONT": Aggregate epr["DF_U64", "DF_U64"], "
				"discard epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_aggregation.epr_lo,
				cont->vc_epr_aggregation.epr_hi,
				epr->epr_lo, epr->epr_hi);
			return -DER_BUSY;
		}

		cont->vc_in_discard = 1;
		cont->vc_epr_discard = *epr;
		break;
	case AGG_MODE_AGGREGATE:
		if (cont->vc_in_aggregation) {
			D_ERROR(DF_CONT": Already in aggregation epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_aggregation.epr_lo, cont->vc_epr_aggregation.epr_hi);
			return -DER_BUSY;
		}

		if (cont->vc_obj_discard_count != 0) {
			D_ERROR(DF_CONT": In object discard epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_discard.epr_lo, cont->vc_epr_discard.epr_hi);
			return -DER_BUSY;
		}

		if (cont->vc_in_discard &&
		    cont->vc_epr_discard.epr_lo <= epr->epr_hi) {
			D_ERROR(DF_CONT": Discard epr["DF_U64", "DF_U64"], "
				"aggregation epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_discard.epr_lo,
				cont->vc_epr_discard.epr_hi,
				epr->epr_lo, epr->epr_hi);
			return -DER_BUSY;
		}

		cont->vc_in_aggregation = 1;
		cont->vc_epr_aggregation = *epr;
		break;
	case AGG_MODE_OBJ_DISCARD:
		if (cont->vc_in_discard) {
			D_ERROR(DF_CONT": In discard epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_discard.epr_lo, cont->vc_epr_discard.epr_hi);
			return -DER_BUSY;
		}

		if (cont->vc_in_aggregation) {
			D_ERROR(DF_CONT": In aggregation epr["DF_U64", "DF_U64"]\n",
				DP_CONT(cont->vc_pool->vp_id, cont->vc_id),
				cont->vc_epr_aggregation.epr_lo, cont->vc_epr_aggregation.epr_hi);
			return -DER_BUSY;
		}

		/** Allow discard from multiple objects */
		cont->vc_obj_discard_count++;
		break;
	}

	return 0;
}

static void
aggregate_exit(struct vos_container *cont, int agg_mode)
{
	switch (agg_mode) {
	default:
		D_ASSERT(0);
		break;
	case AGG_MODE_DISCARD:
		D_ASSERT(cont->vc_in_discard);
		cont->vc_in_discard = 0;
		cont->vc_epr_discard.epr_lo = 0;
		cont->vc_epr_discard.epr_hi = 0;
		break;
	case AGG_MODE_AGGREGATE:
		D_ASSERT(cont->vc_in_aggregation);
		cont->vc_in_aggregation = 0;
		cont->vc_epr_aggregation.epr_lo = 0;
		cont->vc_epr_aggregation.epr_hi = 0;
		break;
	case AGG_MODE_OBJ_DISCARD:
		D_ASSERT(cont->vc_obj_discard_count > 0);
		cont->vc_obj_discard_count--;
		break;
	}
}

static void
merge_window_init(struct agg_merge_window *mw, void (*func)(void *))
{
	struct agg_io_context *io = &mw->mw_io_ctxt;

	memset(mw, 0, sizeof(*mw));
	D_INIT_LIST_HEAD(&mw->mw_phy_ents);
	D_INIT_LIST_HEAD(&mw->mw_phy_rmv_ents);
	D_INIT_LIST_HEAD(&mw->mw_rmv_ents);
	D_INIT_LIST_HEAD(&io->ic_nvme_exts);
	io->ic_csum_recalc_func = func;
}

struct agg_data {
	vos_iter_param_t	ad_iter_param;
	struct vos_agg_param	ad_agg_param;
	struct vos_iter_anchors	ad_anchors;
};

int
vos_aggregate(daos_handle_t coh, daos_epoch_range_t *epr,
	      void (*csum_func)(void *),
	      bool (*yield_func)(void *arg), void *yield_arg, bool full_scan)
{
	struct vos_container	*cont = vos_hdl2cont(coh);
	struct agg_data		*ad;
	int			 rc;

	D_ASSERT(epr != NULL);
	D_ASSERTF(epr->epr_lo < epr->epr_hi && epr->epr_hi != DAOS_EPOCH_MAX,
		  "epr_lo:"DF_U64", epr_hi:"DF_U64"\n",
		  epr->epr_lo, epr->epr_hi);

	D_ALLOC_PTR(ad);
	if (ad == NULL)
		return -DER_NOMEM;

	rc = aggregate_enter(cont, AGG_MODE_AGGREGATE, epr);
	if (rc)
		goto free_agg_data;

	/* Set iteration parameters */
	ad->ad_iter_param.ip_hdl = coh;
	ad->ad_iter_param.ip_epr = *epr;
	/*
	 * Iterate in epoch reserve order for SV tree, so that we can know for
	 * sure the first returned recx in SV tree has highest epoch and can't
	 * be aggregated.
	 */
	ad->ad_iter_param.ip_epc_expr = VOS_IT_EPC_RR;
	/* EV tree iterator returns all sorted logical rectangles */
	ad->ad_iter_param.ip_flags = VOS_IT_PUNCHED | VOS_IT_RECX_COVERED;

	/* Set aggregation parameters */
	ad->ad_agg_param.ap_umm = &cont->vc_pool->vp_umm;
	ad->ad_agg_param.ap_coh = coh;
	ad->ad_agg_param.ap_credits_max = VOS_AGG_CREDITS_MAX;
	ad->ad_agg_param.ap_credits = 0;
	ad->ad_agg_param.ap_discard = 0;
	ad->ad_agg_param.ap_yield_func = yield_func;
	ad->ad_agg_param.ap_yield_arg = yield_arg;
	merge_window_init(&ad->ad_agg_param.ap_window, csum_func);
	/* A full scan caused by snapshot deletion */
	ad->ad_agg_param.ap_full_scan = full_scan;

	ad->ad_iter_param.ip_flags |= VOS_IT_FOR_PURGE;
	rc = vos_iterate(&ad->ad_iter_param, VOS_ITER_OBJ, true, &ad->ad_anchors,
			 vos_aggregate_pre_cb, vos_aggregate_post_cb,
			 &ad->ad_agg_param, NULL);
	if (rc != 0) {
		close_merge_window(&ad->ad_agg_param.ap_window, rc);
		goto exit;
	} else if (ad->ad_agg_param.ap_csum_err) {
		rc = -DER_CSUM;	/* Inform caller the csum error */
		close_merge_window(&ad->ad_agg_param.ap_window, rc);
		/* HAE needs be updated for csum error case */
	}

	/*
	 * Update HAE, when aggregating for snapshot deletion, the
	 * @epr->epr_hi could be smaller than the HAE
	 */
	if (cont->vc_cont_df->cd_hae < epr->epr_hi)
		cont->vc_cont_df->cd_hae = epr->epr_hi;
exit:
	aggregate_exit(cont, AGG_MODE_AGGREGATE);

	if (ad->ad_agg_param.ap_window.mw_csum_support)
		D_FREE(ad->ad_agg_param.ap_window.mw_io_ctxt.ic_csum_buf);

	if (merge_window_status(&ad->ad_agg_param.ap_window) != MW_CLOSED)
		D_ASSERTF(false, "Merge window resource leaked.\n");

free_agg_data:
	D_FREE(ad);

	return rc;
}

int
vos_discard(daos_handle_t coh, daos_unit_oid_t *oidp, daos_epoch_range_t *epr,
	    bool (*yield_func)(void *arg), void *yield_arg)
{
	struct vos_container	*cont = vos_hdl2cont(coh);
	struct agg_data		*ad;
	int			 type = VOS_ITER_OBJ;
	int			 rc;
	int			 mode = oidp == NULL ? AGG_MODE_DISCARD : AGG_MODE_OBJ_DISCARD;

	D_ASSERT(epr != NULL);
	D_ASSERTF(epr->epr_lo <= epr->epr_hi,
		  "epr_lo:"DF_U64", epr_hi:"DF_U64"\n",
		  epr->epr_lo, epr->epr_hi);

	D_ALLOC_PTR(ad);
	if (ad == NULL)
		return -DER_NOMEM;

	rc = aggregate_enter(cont, mode, epr);
	if (rc != 0)
		goto free_agg_data;

	if (oidp != NULL) {
		D_DEBUG(DB_EPC, "Discard "DF_UOID" epr "DF_X64"-"DF_X64"\n", DP_UOID(*oidp),
			epr->epr_lo, epr->epr_hi);
		type = VOS_ITER_DKEY;
		ad->ad_iter_param.ip_oid = *oidp;
		ad->ad_agg_param.ap_discard_obj = 1;
	} else {
		ad->ad_agg_param.ap_discard_obj = 0;
		D_DEBUG(DB_EPC, "Discard epr "DF_X64"-"DF_X64"\n",
			epr->epr_lo, epr->epr_hi);
	}

	/* Set iteration parameters */
	ad->ad_iter_param.ip_hdl = coh;
	ad->ad_iter_param.ip_epr = *epr;
	if (epr->epr_lo == epr->epr_hi)
		ad->ad_iter_param.ip_epc_expr = VOS_IT_EPC_EQ;
	else if (epr->epr_hi != DAOS_EPOCH_MAX)
		ad->ad_iter_param.ip_epc_expr = VOS_IT_EPC_RR;
	else
		ad->ad_iter_param.ip_epc_expr = VOS_IT_EPC_GE;
	/* EV tree iterator returns all sorted logical rectangles */
	ad->ad_iter_param.ip_flags = VOS_IT_PUNCHED | VOS_IT_RECX_COVERED;

	/* Set aggregation parameters */
	ad->ad_agg_param.ap_umm = &cont->vc_pool->vp_umm;
	ad->ad_agg_param.ap_coh = coh;
	ad->ad_agg_param.ap_credits_max = VOS_AGG_CREDITS_MAX;
	ad->ad_agg_param.ap_discard = 1;
	ad->ad_agg_param.ap_credits = 0;
	ad->ad_agg_param.ap_yield_func = yield_func;
	ad->ad_agg_param.ap_yield_arg = yield_arg;

	ad->ad_iter_param.ip_flags |= VOS_IT_FOR_DISCARD;
	rc = vos_iterate(&ad->ad_iter_param, type, true, &ad->ad_anchors,
			 vos_aggregate_pre_cb, vos_aggregate_post_cb,
			 &ad->ad_agg_param, NULL);

	aggregate_exit(cont, mode);

free_agg_data:
	D_FREE(ad);

	return rc;
}
